package com.xrui.hbase.codec.impl;

import com.google.common.base.Preconditions;
import com.moz.kiji.delegation.NamedProvider;
import com.xrui.hbase.RowKeyComponents;
import com.xrui.hbase.codec.RowKeyEncoder;
import com.xrui.hbase.schema.RowKeyComponent;
import com.xrui.hbase.schema.RowKeyEncoding;
import com.xrui.hbase.schema.RowKeyFormat;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

public class RawBytesRowKeyEncoder implements RowKeyEncoder, NamedProvider {
    public static final String PROVIDER_NAME = "RawBytesRowKeyEncoder";

    @Override
    public byte[] encode(RowKeyComponents components, RowKeyFormat format) throws IOException {
        Preconditions.checkArgument(format.getEncoding() == RowKeyEncoding.RAW);
        Preconditions.checkArgument(format.getComponents().size() == 1);
        Preconditions.checkArgument(components.getComponents().length == 1);
        RowKeyComponent component = format.getComponents().get(0);
        Object value = components.getComponents()[0];
        switch (component.getType()) {
            case STRING:
                Preconditions.checkArgument(value instanceof String);
                return Bytes.toBytes((String) value);
            case LONG:
                Preconditions.checkArgument(value instanceof Long);
                return Bytes.toBytes((Long) value);
            case INTEGER:
                Preconditions.checkArgument(value instanceof Integer);
                return Bytes.toBytes((Integer) value);
            default:
                throw new RuntimeException("Not supported row key component type: " + component.getType());
        }
    }

    @Override
    public String getName() {
        return PROVIDER_NAME;
    }
}
